package flink_p1

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 *
 * sessionWindow：
 * 当超过指定时间之后窗口将关闭触发计算并开启一个新窗口，如果一直有数据来则不会触发计算。
 *
 * */

object FlinkTest_18_window_ssession {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)

    socketStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(10)))
      .reduce(new ReduceFunction[(String, Int)] {
        override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {

          (value1._1, value2._2 + value2._2)
        }
      }).print()


    env.execute()
  }

}
